/*
 * Copyright (C) 2018 The Asiainfo-Ability Authors
 *
 *      Licensed under the Apache License, Version 2.0 (the "License");
 *      you may not use this file except in compliance with the License.
 *      You may obtain a copy of the License at
 *
 *          http://www.apache.org/licenses/LICENSE-2.0
 *
 *      Unless required by applicable law or agreed to in writing, software
 *      distributed under the License is distributed on an "AS IS" BASIS,
 *      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *      See the License for the specific language governing permissions and
 *      limitations under the License.
 */

package org.asiainfo.ability.base.service;


import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.asiainfo.ability.base.utils.ExceptionUtils;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateAction;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import javax.annotation.PreDestroy;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;

/**
 * @Author: visen
 * @Date: 2018/1/31
 * @Description:
 */
@SuppressWarnings("unchecked")
public class EsUtilsBean {

    private TransportClient client;
    private BulkProcessor bulkProcessor;
    public static final Logger logger = LogManager.getLogger(EsUtilsBean.class);
    private static final ObjectMapper mapper = new ObjectMapper();

    public EsUtilsBean(String ipAddress, int port) {
        Settings settings = Settings.builder().put("client.transport.sniff", true).build();
        try {
            client = new PreBuiltTransportClient(settings)
                    .addTransportAddress(new TransportAddress(InetAddress.getByName(ipAddress), port));
        } catch (UnknownHostException e) {
            throw new IllegalArgumentException("invalid ip: " + ipAddress);
        }
        bulkProcessor = BulkProcessor.builder(
                client,
                new BulkProcessor.Listener() {

                    @Override
                    public void beforeBulk(long executionId, BulkRequest request) {
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                    }

                })
                .setBulkActions(1000)
                .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                .setConcurrentRequests(4)
                .build();
    }


    public TransportClient getClient() {
        return client;
    }

    private static interface EsAction {
        public void done();
    }

    public static interface AfterSetProp {
        public void put(String key, Object value);
    }

    @PreDestroy
    public void release() {

    }

    public void bulkInsert(final String database, final String table, final Map<String, Object> data) {
        bulkProcessor.add(new IndexRequest(database, table).source(data));
    }

    public void bulkInsert(final String database, final String table, String id, final Map<String, Object> data) {
        bulkProcessor.add(new IndexRequest(database, table, id).source(data));
    }

    public void bulkUpdate(final String database, final String table, String id, final Map<String, Object> data)
            throws Exception {
        UpdateRequest updateRequest = new UpdateRequest();
        updateRequest.index(database);
        updateRequest.type(table);
        updateRequest.id(id);
        XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
        for (Iterator<Map.Entry<String, Object>> it = data.entrySet().iterator(); it.hasNext(); ) {
            Map.Entry<String, Object> entry = it.next();
            builder.field(entry.getKey(), entry.getValue());
        }
        updateRequest.doc(builder.endObject());
        bulkProcessor.add(updateRequest);
    }

    public void bulkDelete(final String database, final String table, String id)
            throws Exception {
        DeleteRequest deleteRequest = new DeleteRequest();
        deleteRequest.index(database);
        deleteRequest.type(table);
        deleteRequest.id(id);
        bulkProcessor.add(deleteRequest);
    }

    public void insert(final String database, final String table, final Map<String, Object> data)
            throws Exception {
        insert(database, table, data, null);
    }

    public void insert(final String database, final String table, final Map<String, Object> data, AfterSetProp prop)
            throws Exception {
        try {
            XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
            for (Iterator<Map.Entry<String, Object>> it = data.entrySet().iterator(); it.hasNext(); ) {
                Map.Entry<String, Object> entry = it.next();
                if (entry.getValue() != null) {
                    builder.field(entry.getKey(), entry.getValue());
                }
            }
            if (prop != null) {

                IndexResponse response = client.prepareIndex(database, table).setSource(builder.endObject()).get();
                /** block */
                if (prop != null) {
                    prop.put("id", response.getId());
                }
            } else {
                /** async */
                client.prepareIndex(database, table).setSource(builder.endObject()).execute();
            }


        } catch (Throwable e) {
            logger.error(e);
        }

    }

    public void update(String database, String table, String id, Map<String, Object> data) throws Exception {
        try {


            //创建修改请求
            UpdateRequest updateRequest = new UpdateRequest();
            updateRequest.index(database);
            updateRequest.type(table);
            updateRequest.id(id);
            XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
            for (Iterator<Map.Entry<String, Object>> it = data.entrySet().iterator(); it.hasNext(); ) {
                Map.Entry<String, Object> entry = it.next();
                builder.field(entry.getKey(), entry.getValue());
            }
            updateRequest.doc(builder.endObject());
            client.execute(UpdateAction.INSTANCE, updateRequest);
            //client.update(updateRequest).get();
        } catch (Throwable e) {
            ExceptionUtils.logTrace(logger, e);
        }


    }

    public void deleteById(String database, String table, String id) {
        DeleteResponse dResponse = client.prepareDelete(database, table, id).execute().actionGet();
    }


    public static class EsColumnMeta {
        private String name;
        private String type;
        private boolean isIndex;

        public EsColumnMeta(String name, String type) {
            this(name, type, true);
        }

        public EsColumnMeta(String name, String type, boolean isIndex) {
            this.name = name;
            this.type = type;
            this.isIndex = isIndex;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public String getType() {
            return type;
        }

        public void setType(String type) {
            this.type = type;
        }

        public boolean isIndex() {
            return isIndex;
        }

        public void setIndex(boolean index) {
            isIndex = index;
        }
    }

    public void createDatabase(String database) {
        CreateIndexRequest request = new CreateIndexRequest(database);
        ActionFuture future = client.admin().indices().create(request);
        try {
            future.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void createTable(String database, String table, List<EsColumnMeta> metaData) throws Exception {
        try {
            XContentBuilder xContentBuilder = XContentFactory.jsonBuilder().startObject()
                    .startObject("properties");
            for (EsColumnMeta item : metaData) {
                xContentBuilder.startObject(item.getName())
                        .field("type", item.getType());
                if (!item.isIndex()) {
                    xContentBuilder.field("index", false);
                }
                xContentBuilder.endObject();
            }
            xContentBuilder.endObject().endObject();
            PutMappingRequest mapping = Requests.putMappingRequest(database).type(table).source(xContentBuilder);
            client.admin().indices().putMapping(mapping).actionGet();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    public boolean isExistsDatabase(String database) {
        IndicesExistsResponse response =
                client.admin().indices().exists(
                        new IndicesExistsRequest().indices(new String[]{database})).actionGet();
        return response.isExists();
    }

    public boolean isExistsTable(String database, String table) {
        TypesExistsResponse response = client.admin().indices().typesExists(
                new TypesExistsRequest(new String[]{database}, table)).actionGet();
        return response.isExists();
    }
    public static void main(String[] args) throws Exception {
        EsUtilsBean es = new EsUtilsBean("192.168.17.129", 9300);
        //es.deleteDatabase("machine_status_db");
        //es.deleteDatabase("record_db");
        //es.deleteDatabase("test_db");
        //es.deleteDatabase("trace_callee_db");
        //es.deleteDatabase("trace_db");
        //es.deleteDatabase("trace_caller_db");
        es.deleteById("trace_caller_db","trace_caller", "vsO5nmIB6KU-KIxLm9wS");
        es.deleteById("trace_caller_db","trace_caller", "vcO4nmIB6KU-KIxL8dxh");
    }
    public static void main1(String[] args) throws Exception {
        EsUtilsBean es = new EsUtilsBean("192.168.17.129", 9300);
        if (!es.isExistsDatabase("machine_status_db")) {
            es.createDatabase("machine_status_db");
        }

        if (!es.isExistsTable("machine_status_db", "machine_status")) {
            List<EsUtilsBean.EsColumnMeta> metas = new ArrayList();
            metas.add(new EsUtilsBean.EsColumnMeta("machineid", "text"));
            metas.add(new EsUtilsBean.EsColumnMeta("cpuRatio", "text"));
            metas.add(new EsUtilsBean.EsColumnMeta("memRatio", "text"));
            metas.add(new EsUtilsBean.EsColumnMeta("time", "long"));
            try {
                es.createTable("machine_status_db", "machine_status", metas);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        Map param = new HashMap();
        param.put("machineid", "1");
        param.put("cpuRatio", "2");
        param.put("memRatio", "3");
        param.put("time", System.currentTimeMillis());
        es.bulkInsert("machine_status_db", "machine_status", "1", param);
        Thread.sleep(10000);
        // es.deleteDatabase("d");
        //es.deleteDatabase("record_db");
        //es.deleteDatabase("trace_db");
        //es.deleteDatabase("trace_summary_db");
        if (!es.isExistsDatabase("test_db")) {
            es.createDatabase("test_db");
        }

        if (!es.isExistsTable("test_db", "test")) {
            List<EsUtilsBean.EsColumnMeta> metas = new ArrayList();
            metas.add(new EsUtilsBean.EsColumnMeta("traceid", "text"));
            try {
                es.createTable("test_db", "test", metas);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        StringBuilder sb = new StringBuilder();
        sb.append("abc");
        sb.append("321");
        sb.deleteCharAt(sb.length() - 1);
        System.out.println(sb.toString());
        long old = System.currentTimeMillis();
        Map data = new HashMap();
        data.put("traceid", "2183042439018250240 2183124913790451713");
        System.out.println(es.search("record_db", "record", data, null, null));
        BulkProcessor b = BulkProcessor.builder(
                es.getClient(),
                new BulkProcessor.Listener() {

                    @Override
                    public void beforeBulk(long executionId, BulkRequest request) {
                        logger.info("---尝试插入{}条数据---", request.numberOfActions());
                    }

                    @Override
                    public void afterBulk(long executionId,
                                          BulkRequest request, BulkResponse response) {
                        logger.info("---尝试插入{}条数据成功---", request.numberOfActions());
                    }

                    @Override
                    public void afterBulk(long executionId,
                                          BulkRequest request, Throwable failure) {
                        logger.error("[es错误]---尝试插入数据失败---", failure);
                    }

                })
                .setBulkActions(1000)
                .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                .setConcurrentRequests(10)
                .build();
        Map d = new HashMap();
        d.put("traceid", "333");


        //b.add(new IndexRequest("test_db", "test", "444").source(d));
        old = System.currentTimeMillis();


        for (int i = 0; i < 10000; i++) {
            d.put("traceid", "333" + i);
            //es.insert("test_db", "test",d);
            //b.add(new IndexRequest("test_db", "test").source(d));
        }
        d = new HashMap();
        d.put("traceid", "2184859319345414150 2184837447740358656");

        System.out.println(es.search("record_db", "record", d));


        System.out.println(System.currentTimeMillis() - old);

    }

    public static class Page {
        private String orderBy;
        private SortOrder order = SortOrder.ASC;
        private int curPage = 0;
        private int pageCount = 10;

        public Page(String orderBy) {
            this.orderBy = orderBy;
        }

        public Page(String orderBy, SortOrder order) {
            this.orderBy = orderBy;
            this.order = order;
        }

        public String getOrderBy() {
            return orderBy;
        }

        public void setOrderBy(String orderBy) {
            this.orderBy = orderBy;
        }

        public SortOrder getOrder() {
            return order;
        }

        public void setOrder(SortOrder order) {
            this.order = order;
        }

        public int getCurPage() {
            return curPage;
        }

        public void setCurPage(int curPage) {
            this.curPage = curPage;
        }

        public int getPageCount() {
            return pageCount;
        }

        public void setPageCount(int pageCount) {
            this.pageCount = pageCount;
        }
    }

    public List<Map> search(String database, String table, Map<String, Object> condition) {
        return search(database, table, condition, null, null);
    }

    public List<Map> search(String database, String table, Map<String, Object> conditions, List fetchSources
            , Page page) {
        List<QueryBuilder> condList = new ArrayList();
        for (Iterator<Map.Entry<String, Object>> it = conditions.entrySet().iterator(); it.hasNext(); ) {
            Map.Entry<String, Object> entry = it.next();
            if (entry.getValue() instanceof QueryBuilder) {
                condList.add((QueryBuilder) entry.getValue());
            } else {
                condList.add(QueryBuilders.matchQuery(entry.getKey(), entry.getValue()));
            }

        }

        return search(database, table, condList, fetchSources, null);
    }

    public List<Map> search(String database, String table, List<QueryBuilder> conditions, List<String> fetchSources
            , Page page) {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        for (QueryBuilder cond : conditions) {
            boolQueryBuilder.must(cond);
        }
        SearchResponse response = null;

        try {
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            SearchRequestBuilder builder = client.prepareSearch(database).setTypes(table);
            if (fetchSources != null && !fetchSources.isEmpty()) {
                searchSourceBuilder.fetchSource(fetchSources.toArray(new String[]{}), null);
                builder.setSource(searchSourceBuilder);
            }
            builder.setQuery(boolQueryBuilder);
            if (page != null) {
                builder.setFrom(page.getCurPage() * page.getPageCount()).setSize(page.getPageCount());
                if (page.getOrder() != null) {
                    builder.addSort(page.getOrderBy(), page.getOrder());
                }
            } else {
                builder.setFrom(0).setSize(10);
            }
            builder.setExplain(false);
            response = builder.execute().get();
        } catch (Exception e) {
            e.printStackTrace();
        }
        List<Map> result = new ArrayList<>();
        if (result != null) {
            for (SearchHit searchHit : response.getHits()) {
                result.add(searchHit.getSourceAsMap());
            }
        }


        return result;
    }

    public void search1(String database, String table, Map<String, String> conditions) {

        QueryBuilder qb = QueryBuilders.termQuery(
                "info.keyword",
                "<abc>123-4567"
        );
        QueryBuilder qb2 = QueryBuilders.matchQuery(
                "info",
                "<abc>1"
        );
        /*
        QueryBuilder queryBuilder1 = QueryBuilders.wildcardQuery("info", "*123*5*");
        SearchResponse response = client.prepareSearch(database).setQuery(queryBuilder1)
                .setTypes(table).get();
        for (SearchHit  searchHit: response.getHits()) {
            println(searchHit);
        }
*/
        // QueryBuilders.queryStringQuery()
        QueryBuilder queryBuilder = QueryBuilders.rangeQuery("traceid")
                .from("1")
                .to("11");
        QueryBuilder qb1 = QueryBuilders.termQuery(
                "traceid",
                "222"

        );

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery()
                .must(qb2);
        //.must(queryBuilder);

        SearchResponse response = null;
        try {
            response = client.prepareSearch(database).setTypes(table)
                    .setQuery(boolQueryBuilder)
                    .setFrom(0).setSize(500)
                    .setExplain(false)

                    //.addStoredField(new String[]{"cphm1","jcdid","cplx1","tpid1","tgsj","cdid"})
                    .addSort("time", SortOrder.DESC).execute()
                    //.addAggregation(AggregationBuilders.avg("time1").field("time")).execute()
                    .get();
            //InternalAvg sum=response.getAggregations().get("time1");
            //System.out.println(sum.getValue());
        } catch (Exception e) {
            e.printStackTrace();
        }
        for (SearchHit searchHit : response.getHits()) {
            println(searchHit);
        }


    }

    public static void println(SearchHit searchHit) {
        System.err.println("*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-");
        System.err.println(
                "docId : " + searchHit.docId() + "\n" +
                        "getId : " + searchHit.getId() + "\n" +
                        "getIndex : " + searchHit.getIndex() + "\n" +
                        "getScore : " + searchHit.getScore() + "\n" +
                        "getSourceAsString : " + searchHit.getSourceAsString() + "\n" +
                        "getType : " + searchHit.getType() + "\n" +
                        "getVersion : " + searchHit.getVersion() + "\n" +
                        "fieldsOrNull : " + searchHit.fieldsOrNull() + "\n" +
                        "getExplanation : " + searchHit.getExplanation() + "\n" +
                        "getFields : " + searchHit.getFields() + "\n" +
                        "hasSource : " + searchHit.hasSource()
        );
    }

    public void createIndex(String database, String table) throws Exception {

        CreateIndexRequestBuilder cib = client.admin().indices().prepareCreate(database);
        XContentBuilder mapping = XContentFactory.jsonBuilder()
                .startObject()
                .startObject("properties") //设置之定义字段
                .startObject("traceid")
                .field("type", "string") //设置数据类型
                .endObject()
                .startObject("bizid")
                .field("type", "string")
                .endObject()
                .startObject("info")
                .field("type", "string")
                .endObject()
                .startObject("time")
                .field("type", "long")
                .endObject()
                .endObject()
                .endObject();
        cib.addMapping(table, mapping);

        CreateIndexResponse res = cib.execute().actionGet();

        System.out.println("----------添加映射成功----------");
    }

    public void deleteDatabase(String database) {
        DeleteIndexResponse dResponse = client.admin().indices().prepareDelete(database)
                .execute().actionGet();
        System.out.println("是否删除成功:" + dResponse.isAcknowledged());
    }


}
